package com.wanke;

import cn.beepower.domain.bus.MqttDataBus;
import cn.beepower.domain.kit.Log;
import cn.beepower.domain.mqtt.client.MQTT;
import com.wanke.scada.utils.PropertiesLoadUtils;
import com.wanke.config.HttpConfig;
import com.wanke.config.MeasurementsCodec;
import com.wanke.constant.ProtocolConstants;
import com.wanke.domain.po.Measurements;
import io.vertx.core.Vertx;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.wanke.subscribe.MeasureTransmitterSubscribe;
import com.wanke.subscribe.MqttMeasuremenSubscribetRecevice;
import com.wanke.service.impl.MeasurementServiceImpl;
import com.wanke.publish.HttpRemoteMeasurePublish;
import com.wanke.publish.MqttRemoteMeasurePublish;
import com.wanke.publish.RemoteMeasurePublish;
import com.wanke.publish.WebsocketRemoteMeasurePublish;
import com.wanke.util.InfluxDBConnection;
import com.wanke.util.InfluxDBUtil;

import java.net.URISyntaxException;
import java.util.Objects;
import java.util.Properties;

/**
 * 数据上传程序
 * @author diaoshi
 * @date 2021-01-19
 */
public class TransmitterMain {

    private static final Logger logger = LoggerFactory.getLogger(TransmitterMain.class);

    public static Properties properties;

    public static void main(String[] args) {
        properties = PropertiesLoadUtils.readProperties("application");
        String protocol = properties.get("update.protocol").toString();
        String url = properties.get("influxdb.dbUrl").toString();
        String databaseUserName = properties.get("influxdb.username").toString();
        String databasePassword = properties.get("influxdb.password").toString();

        Long stationCode = Long.valueOf(properties.get("station.code").toString());
        Long frequency = Long.valueOf(properties.get("custom.collect.heartbeat.frequency").toString());
        Long timeOut = Long.valueOf(properties.get("custom.collect.heartbeat.timeout").toString());
        Long timeOutPoint = Long.valueOf(properties.get("transfer.data.timeout.pointNumber").toString());
        Integer pollSize = Integer.valueOf(properties.get("transfer.data.maxSize").toString());

        MeasurementServiceImpl measurementService = new MeasurementServiceImpl(new InfluxDBUtil(new InfluxDBConnection(databaseUserName,databasePassword,url,null)));

        String mqtt_userName = properties.get("mqtt.username").toString();
        String mqtt_password = properties.get("mqtt.password").toString();
        String mqttHost = properties.get("mqtt_server_host").toString();
        String mqttClientId =properties.get("local.clientId").toString();
        MqttDataBus mqttDataBus =  mqttDataBus(mqtt_userName,mqtt_password,mqttHost ,mqttClientId);
        String beeId = properties.get("local_bee_id").toString();
        MqttMeasuremenSubscribetRecevice mqttMeasurementRecevice = new MqttMeasuremenSubscribetRecevice(mqttDataBus, stationCode,beeId);

        Vertx vertx = Vertx.vertx();
        vertx.eventBus().registerDefaultCodec(Measurements.class, new MeasurementsCodec());
        vertx.deployVerticle(mqttMeasurementRecevice,r -> {
            if (r.succeeded()) {
                logger.error("mqtt量测订阅启动成功");
            } else {
                logger.error("mqtt量测订阅启动失败",r.cause());
            }
        });

        MeasureTransmitterSubscribe transmitterPublish = new MeasureTransmitterSubscribe(1, 5000,5000);
        vertx.deployVerticle(transmitterPublish,r -> {
            if (r.succeeded()) {
                logger.error("转发器启动成功");
            } else {
                logger.error("转发器启动失败",r.cause());
            }
        });

        if (Objects.equals(protocol, ProtocolConstants.MQTT.getProtocol())) {
            String remoteClientId =  properties.get("remote.clientId").toString();
            MqttRemoteMeasurePublish  remoteMeasureUpload = new MqttRemoteMeasurePublish(mqttDataBus,
                    measurementService,
                    stationCode, timeOut, pollSize, frequency, remoteClientId, timeOutPoint);
            vertx.deployVerticle(remoteMeasureUpload, r -> {
                if (r.succeeded()) {
                    logger.error("mqtt发布器启动成功");
                } else {
                    logger.error("mqtt发布器启动失败",r.cause());
                }
            });
         }
        String userName = properties.get("http.username").toString();
        String password = properties.get("http.password").toString();
        String type = properties.get("http.type").toString();
        String httpHost = properties.get("http.host").toString();
        if (Objects.equals(protocol, ProtocolConstants.HTTP.getProtocol())) {
            HttpConfig.init(vertx,httpHost);
            RemoteMeasurePublish remoteMeasureUpload = new HttpRemoteMeasurePublish(measurementService,
                    userName, password,type,timeOut, timeOutPoint,stationCode,frequency, pollSize, httpHost);
            vertx.deployVerticle(remoteMeasureUpload, r -> {
                if (r.succeeded()) {
                    logger.error("http发布器启动成功");
                } else {
                    logger.error("http发布器启动失败",r.cause());
                }
            });
        }

        if (Objects.equals(protocol, ProtocolConstants.WEBSOCKET.getProtocol())) {
            String host = properties.get("websocket.host").toString();
            int port = Integer.valueOf(properties.get("websocket.port").toString());
            HttpConfig.init(vertx,httpHost);
             RemoteMeasurePublish remoteMeasureUpload = new WebsocketRemoteMeasurePublish(measurementService,port,stationCode,host,
                    userName, password,type,timeOutPoint,timeOut,frequency, pollSize, httpHost);
            vertx.deployVerticle(remoteMeasureUpload, r -> {
                if (r.succeeded()) {
                    logger.error("websocket发布器启动成功");
                } else {
                    logger.error("websocket发布器启动失败",r.cause());
                }
            });
        }
    }

    public static MqttDataBus mqttDataBus(String username, String password, String host, String localId) {
        MQTT mqtt = new MQTT();
        mqtt.setUserName(username);
        mqtt.setPassword(password);
        try {
            mqtt.setHost(host);
        } catch (URISyntaxException e) {
            logger.error("错误的mqtt Url", e);
        }
        mqtt.setClientId(localId);
        Log.INSTANCE.setFilter("bus:I gate:E fes:E *:E");
        MqttDataBus mqttDataBus = new MqttDataBus(mqtt);
        mqttDataBus.publish("test");
        return mqttDataBus;
    }
}
